package net.wlm.jtp.dws.log.function;


import net.wlm.jtp.dws.log.bean.PageViewBean;
import net.wlm.jtp.dws.log.utils.DateTimeUtil;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class PageViewReportWindowFunction implements WindowFunction<PageViewBean, String, String, TimeWindow> {
    public void apply(String key, TimeWindow window, Iterable<PageViewBean> input, Collector<String> out) throws Exception{
         /*
       key：表示分组字段；window：表示时间窗口，可以获取窗口开始和结束时间；input：表示窗口中所有数据
        */
        // 获取窗口时间
        String windowStartTime = DateTimeUtil.convertLongToString(window.getStart(), DateTimeUtil.DATE_TIME_FORMAT);
        String windowEndTime = DateTimeUtil.convertLongToString(window.getEnd(), DateTimeUtil.DATE_TIME_FORMAT);

        // 获取窗口计算结果
        PageViewBean pageViewBean = input.iterator().next();
        Long pvCount = pageViewBean.getPvCount();
        Long uvCount = pageViewBean.getUvCount();
        Long pvDuringTime = pageViewBean.getPvDuringTime();
        Long sessionCount = pageViewBean.getSessionCount();

        // 输出
        String output = windowStartTime + "," +  windowEndTime + "," +
                key + "," +
                pvCount + "," + pvDuringTime + "," +  uvCount + "," +   sessionCount +","
                + System.currentTimeMillis();
        out.collect(output);
    }


}
